package com.wang.event;

import java.util.*;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.function.Consumer;

public class SelectorEventManager<T> {
    private List<Consumer<Event<T>>> consumerList = new ArrayList<>();
    private Map<Consumer<Event<T>>, BlockingQueue<Event<T>>> map = new HashMap<>();
    private Executor executor;
    // 事件回调机制
    private BlockingQueue<Consumer<Event<T>>> successCallback = new LinkedBlockingQueue<>();

    public SelectorEventManager(Executor executor) {
        this.executor = executor;
        // 启动selector监听
        new Thread(() -> {
            try {
                while (true) {
                    Consumer<Event<T>> consumer = successCallback.take();
                    Event<T> event = map.get(consumer).take();
                    this.executor.execute(() -> {
                        consumer.accept(event);
                        this.successCallback.add(consumer);
                    });
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();
    }

    public void register(Consumer<Event<T>> consumer) {
        consumerList.add(consumer);
    }

    public void sendEvent(Event<T> event) {
        map.values().forEach(v -> v.add(event));
        for (Consumer<Event<T>> eventConsumer : consumerList) {
            this.executor.execute(() -> {
                eventConsumer.accept(event);
            });
        }
    }

    private final BlockingQueue<String> queue = new LinkedBlockingQueue<>();
    private volatile boolean isRunning = false;

    public void add(String s) {
        this.queue.add(s);
        synchronized (this) {
            if (isRunning) {
                return;
            }
        }
        this.run();
    }

    public void run() {
        String s;
        while ((s = queue.poll()) != null) {
            System.out.println(s);
        }
        synchronized (this) {
            s = queue.poll();
            if (s == null) {
                this.isRunning = false;
                return;
            }
        }
        this.run();
    }

    public void newRun() {
        Runnable r = () -> {
            String s;
            while ((s = queue.poll()) != null) {
                System.out.println(s);
            }
        };
        r.run();
        synchronized (this) {
            if (queue.element() == null) {
                this.isRunning = false;
                return;
            }
        }
        r.run();
    }
}
